package sc.analysis.vortex;


import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import sc.writable.IntermediateData;
import sc.writable.IntermediateData.Type;


/**
 * Reducer of the Vortex Indicator.
 *
 * @author Quentin Rossé
 *
 */
public class VortexReducer extends Reducer<Text, VortexData, 
                                        Text, IntermediateData> {
	
	private static final String ANALYSIS_NAME = "Vortex";
	
	/**
	 * data written for the next phase
	 */
	private static IntermediateData imd = new IntermediateData();
	
	@Override
	public void reduce(Text key, Iterable<VortexData> values, Context context) throws IOException, InterruptedException {
		System.out.println(key);
		for(VortexData vd : values){
			System.out.println("VORTEX"+ vd.getPeriod() + "   UP : "+vd.getUp() + 
					"  DOWN : "+vd.getDown() + "  DSC : " + 
					vd.getDaysSinceCross() + "  delta : " + 
					Math.abs(vd.getUp() - vd.getDown()) + "  score : " + 
					score(vd.getUp(), vd.getDown(), vd.getDaysSinceCross()));
					
			imd.setAnalysisName(ANALYSIS_NAME + vd.getPeriod());
			imd.setScore(score(vd.getUp(), vd.getDown(), vd.getDaysSinceCross()));
			imd.setType(vd.getPeriod());
			if(vd.getDaysSinceCross() != -1)
				imd.setDetails("days since cross : "+vd.getDaysSinceCross());
			else
				imd.setDetails("");
			context.write(key, imd);
			
			
			if(vd.getPeriod() == 0){
				imd.setScore(1);
				if(vd.getUp() > vd.getDown()){
					imd.setType(Type.ALERT_BULL);
					imd.setDetails("VortexUp crossed VotexDown");
				}
				else{
					imd.setType(Type.ALERT_BEAR);
					imd.setDetails("VortexDown crossed VotexUp");
				}
				context.write(key, imd);
			}
		}
	}
	
	@Override
	public void setup(Context context){
		imd.setAnalysisName(ANALYSIS_NAME);
	}
	
	@Override
	public void cleanup(Context context){}
	
	/**
	 * Return the score from the value of VortexUp and VortexDown
	 * 
	 * @return
	 */
	private int score(float vortexUp, float vortexDown, int daysSinceCross){
		int tmpScore;
		float delta = Math.abs(vortexUp - vortexDown);
		
		//score for number of days since cross, the smaller the better (if it's
		//too long, the opportunity has gone)
		if(daysSinceCross < 0)
			tmpScore = 0;
		else if(daysSinceCross <= 3)
			tmpScore = 5;
		else if(daysSinceCross <= 10)
			tmpScore = 4;
		else if(daysSinceCross <= 20)
			tmpScore = 3;
		else if(daysSinceCross <= 50)
			tmpScore = 2;
		else
			tmpScore = 1;
		
		if(delta >= 0.02){ //else it's +0
			if(delta <= 0.07)
				tmpScore += 1;
			else if(delta <= 0.15)
				tmpScore += 2;
			else if(delta <= 0.3)
				tmpScore += 3;
			else if(delta <= 0.8)
				tmpScore += 4;
			else
				tmpScore += 5;
		}
			
		return vortexUp>vortexDown ? (5+(tmpScore+1)/2) : (5-tmpScore/2);
	}
}